// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: fengyang, gekim

#include "coin2/exchange/bybit_futures/feed_v3/processor.h"

#include <string>
#include <utility>

#include "coin2/base/conversion.h"
#include "coin2/base/string_util.h"
#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/bybit_futures/feed_v3/update.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::bybit_futures::feed_v3::impl {
namespace {
void ParseTickers(
    const rapidjson::Value& ticker_json, int64_t ts,
    base::feed::impl::FeedUpdateManager* upd_mgr,
    const std::string& native_symbol,
    bool is_snapshot,
    bool open_interest_only) {
  if (ticker_json.HasMember("openInterest")) {
    auto* open_interest = upd_mgr->GetOpenInterestMutable();
    BybitFeedUpdateParser::ParseOpenInterest(ticker_json, open_interest, ts);
    upd_mgr->PublishOpenInterestWithExchangeTimeCheck(native_symbol);
  }
  if (!open_interest_only) {
    if (ticker_json.HasMember("markPrice")) {
      auto* mark_price = upd_mgr->GetMarkPriceMutable();
      BybitFeedUpdateParser::ParseMarkPrice(ticker_json, mark_price, ts);
      upd_mgr->PublishMarkPriceWithExchangeTimeCheck(native_symbol);
    }
    if (ticker_json.HasMember("indexPrice")) {
      auto* index = upd_mgr->GetIndexMutable();
      BybitFeedUpdateParser::ParseIndex(ticker_json, index, ts);
      upd_mgr->PublishIndexWithExchangeTimeCheck(native_symbol);
    }
    if (ticker_json.HasMember("fundingRate") ||
        ticker_json.HasMember("nextFundingTime")) {
      auto* funding_rate = upd_mgr->GetFundingRateMutable();
      if (BybitFeedUpdateParser::ParseFundingRate(ticker_json, funding_rate, ts, is_snapshot)) {
        upd_mgr->PublishFundingRateWithExchangeTimeCheck(native_symbol);
      }
    }
  }
}
}  // namespace

void BybitFeedProcessor::OnTopicRecord(
    const base::executor::TopicRecordPacket& packet) {
  upd_mgr_.onBeginPacket(packet);

  rapidjson::Document document;
  document.Parse<rapidjson::kParseNumbersAsStringsFlag>(packet.payload.data());
  if (document.HasParseError()) {
    throw std::runtime_error(std::string("JSON parse error: ") +
                             GetParseError_En(document.GetParseError()));
  }

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(document);
  }

  if ((document.HasMember("ret_msg") && document["ret_msg"] == "pong") ||
      (document.HasMember("ret_msg") && document.HasMember("op") &&
       document["op"] == "subscribe")) {
    upd_mgr_.onEndPacket(packet);
    return;
  }

  BybitChannel channel;
  std::string native_symbol;
  fastfeed::proto::KlineInterval interval;
  int page = -1;

  if (document.HasMember("topic")) {
    const std::string topic = document["topic"].GetString();
    channel = BybitChannel::Parse(topic);
    native_symbol = channel.symbol;
  } else if (document.HasMember("url")) {
    if (BybitFeedUpdateParser::ParseKlineUrl(
        document["url"].GetString(), &native_symbol, &interval, &page)) {
      channel = {"kline_rest", native_symbol};
    } else {
      upd_mgr_.onEndPacket(packet);
      return;
    }
  } else {
    LOG(INFO) << "[Bybit Feed Processor] meta msg for " << packet.topic_name
              << ": " << packet.payload.data();
    upd_mgr_.onEndPacket(packet);
    return;
  }

  if (!CheckChannelFilter(channel.type)) {
    upd_mgr_.onEndPacket(packet);
    return;
  }

  if (!upd_mgr_.is_symbol_registered(native_symbol)) {
    upd_mgr_.onEndPacket(packet);
    return;
  }

  if (channel.type == "orderbook.1" || channel.type == "orderbook.50") {
    int64_t ts = stoll(document["ts"].GetString()) * 1'000'000LL;
    auto* book = upd_mgr_.GetBookMutable(native_symbol);
    if (parser_config_.publish_bbo_book_only()) {
      std::optional<PriceQty> bid_pre = book->Bid0();
      std::optional<PriceQty> ask_pre = book->Ask0();
      if (BybitFeedUpdateParser::ParseBook(
              book, ts, document, true, native_symbol, packet.timestamp)) {
        std::optional<PriceQty> bid_post = book->Bid0();
        std::optional<PriceQty> ask_post = book->Ask0();
        bool is_bbo_changed = (bid_pre != bid_post) || (ask_pre != ask_post);
        if (is_bbo_changed) {
          upd_mgr_.PublishBook();
        }
      }
    } else if (parser_config_.has_sampling_frequency_bps()) {
      if (BybitFeedUpdateParser::ParseBook(
              book, ts, document, true, native_symbol, packet.timestamp)) {
        bool sampled = book->SampleByFrequency(
              book->Bid0(), book->Ask0(),
              parser_config_.sampling_frequency_bps());
        if (parser_config_.has_trade_sampling_frequency_bps()) {
          auto* trade = upd_mgr_.GetTradeMutable();
          auto* slim_trade = upd_mgr_.GetSlimTradeMutable(native_symbol);
          int64_t timespan = packet.timestamp - slim_trade->LastPublishedTimestamp();
          if (sampled ||
             timespan > parser_config_.trade_sampling_time_length()) {
            if (slim_trade->PopulateTradeAndClear(trade)) {
              upd_mgr_.PublishTradeWithExchangeTimeCheck(native_symbol, false);
            }
          }
        }

        if (sampled) {
          upd_mgr_.PublishBook();
        }
      }
    } else {
      if (BybitFeedUpdateParser::ParseBook(
              book, ts, document, false, native_symbol, packet.timestamp)) {
        upd_mgr_.PublishBook();
      }
    }
  } else if (channel.type == "publicTrade") {
    if (parser_config_.has_trade_sampling_frequency_bps()) {
      auto* slim_trade = upd_mgr_.GetSlimTradeMutable(native_symbol);
      for (const auto& trade_json : document["data"].GetArray()) {
        Trade _trade{};
        BybitFeedUpdateParser::ParseTrade(trade_json, &_trade);
        bool aggregated = slim_trade->TryAggregateSlimTrade(
            packet.timestamp,
            _trade.price,
            _trade.fill_qty,
            _trade.side,
            _trade.timestamp,
            parser_config_.trade_sampling_frequency_bps());
        int64_t timespan = packet.timestamp - slim_trade->LastPublishedTimestamp();
        if (!aggregated ||
            timespan > parser_config_.trade_sampling_time_length()) {
          auto* trade = upd_mgr_.GetTradeMutable();
          CHECK(slim_trade->PopulateTradeAndClear(trade));
          upd_mgr_.PublishTradeWithExchangeTimeCheck(native_symbol, false);
        }
      }
    } else {
      Trade* agg_trade = upd_mgr_.GetTradeMutable();
      agg_trade->Clear();
      Trade trade{};

      for (auto& trade_json : document["data"].GetArray()) {
        BybitFeedUpdateParser::ParseTrade(trade_json, &trade);
        if (TryAggregateTrades(agg_trade, trade)) {
          continue;
        }
        if (agg_trade->fill_qty > 0) {
          upd_mgr_.PublishTradeWithExchangeTimeCheck(native_symbol, true);
        }
        *agg_trade = trade;
      }
      if (agg_trade->fill_qty > 0) {
        upd_mgr_.PublishTradeWithExchangeTimeCheck(native_symbol, false);
      }
    }
  } else if (channel.type == "tickers") {
    int64_t ts = stoll(document["ts"].GetString()) * 1'000'000LL;
    ParseTickers(document["data"], ts, &upd_mgr_, native_symbol, true, open_interest_only_);
  } else if (channel.type == "kline_rest") {
    auto* kline = upd_mgr_.GetKlineMutable(native_symbol, interval);
    if (kline != nullptr) {
      bool to_publish = false;
      // query is split up several requests
      if (page == 0) {
        // first long query
        kline->SnapshotClear();
      } else {
        kline->UpdateClear();
      }

      const auto& items = document["data"]["result"]["list"].GetArray();
      for (int i = items.Size() - 1; i >= 0; i--) {
        bool freezed = BybitFeedUpdateParser::Parse1Kline(
            items[i], kline, packet.timestamp, interval);
        to_publish = to_publish || freezed;
      }

      if (to_publish) {
        upd_mgr_.PublishKline(native_symbol);
      }
    }
  } else {
    if (parser_config_.continue_on_error()) {
      LOG(ERROR) << "[Bybit Feed Processor] unkown topic: " << packet.payload;
    } else {
      THROW() << "[Bybit Feed Processor] unkown topic: " << packet.payload;
    }
  }

  upd_mgr_.onEndPacket(packet);
}
}  // namespace coin2::exchange::bybit_futures::feed_v3::impl
